package com.heima.kafka.sample;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 流式处理
 */
public class KafkaStreamQuickStart {
    public static void main(String[] args) {

        //kafka的配置信息
        Properties properties = new Properties();
        //kafka集群地址  需要到虚拟机上启动kafka服务
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        //(stream相关的序列化)key的序列化
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        //(stream相关的序列化)value的序列化
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        //应用的名称
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"stream-quick-start");

        //stream构建器
        StreamsBuilder StreamBuilder = new StreamsBuilder();

        //流式计算
        streamProcessor(StreamBuilder);

        //创建kafkaStream生产者对象                    拿到拓扑图(指连接结构)？？？
        KafkaStreams kafkaStreams = new KafkaStreams(StreamBuilder.build(),properties);
        //启动流式计算
        kafkaStreams.start();

    }

    /**
     * 流式计算
     * 消息内容：hello kafka(就是下面的value) hello itcast
     * @param StreamBuilder
     */
    private static void streamProcessor(StreamsBuilder StreamBuilder) {
        //1.接收消息(来自隔壁的ProducerQuickStart.java里的new ProducerRecord<String,String>("itcast-topic-input","hello kafka");)
        // 创建kstream对象，同时指定从哪个消息的生产者(topic)中去拿到消息
        KStream<String, String> stream = StreamBuilder.stream("itcast-topic-input");
        //处理当前的流式计算的时候，都是string这种字符串类型，所以把泛型Iterable<?>都改成Iterable<String>
        /**
         * 处理消息的value
         * @param value
         * @return
         */
        //2.聚合处理(把当前的每个词都放进集合Iterable中(作为消息的value))
        stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override //list集合继承了Iterable(集合)
            public Iterable<String> apply(String value) {
                //valAry就是某一个单词
                String[] valAry = value.split(" ");
                //转成集合
                return Arrays.asList(valAry);//返回集合，里面应该就包含某一个单词
                //缩写：
                //return Arrays.asList(value.split(" "));
            }
            //链式编程，继续往下面调用方法
            //处理之后value就是当前某个集合，这个集合就是当前某个单词，
            // 所以目前还是按照value进行聚合
        }).groupBy((key, value)-> value)
                //时间窗口设置，处理当前流逝的时候，每隔10秒，统计(聚合)一次数据，把当前10秒内的数据进行聚合，然后把聚合的结果放到一个topic中
                //在10秒内点击启动ProducerQuickStart两次，就会统计发送的两倍信息
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10))) //不推荐使用：TimeWindows.of(10000)
                //按照当前的value进行统计单词个数
                .count()
                //把结果放到一个topic中，topic名称：stream-quick-output
                //转换为kStream对象
                .toStream()
                //TODO ???一旦处理完成，当前消息的key就变了(变成了某一个单词)，value就变成了count之后的一个个数，
                // 为了方便查看，要做key和value的转换(转成字符串)
                //转换为字符串输出
                .map((key, value)->{
                    //需求：求单词的个数
                    //结果：
//                    key:[hello@1729074830000/1729074840000] ,value:10
//                    key:[kafka@1729074830000/1729074840000] ,value:10
                    System.out.println("key:"+key+" ,value:"+value);
                    return new KeyValue<>(key.key().toString(),value.toString());
                })
                //发送消息(隔壁的ConsumerQuickStart.java里的consumer.subscribe(Collections.singletonList("itcast-topic-out"));接收)
                .to("itcast-topic-out");
    }
}
